package connect.client;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import connect.Decoder;
import connect.Encoder;
import connect.Message;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

public class CommunicationManager {
	
	private Client client;//ip,端口
	private ResponseListener listener;//消息回调监听
	private ChannelHandlerContext context;//消息发送器
	private EventLoopGroup group;
	private ExecutorService pool;

	/**
	 * 构造
	 * @param client
	 * @param listener
	 */
	public CommunicationManager(final Client client,final ResponseListener listener) {
		pool=Executors.newCachedThreadPool();
		this.client=client;
		this.listener=listener;
		reConnect(null);
	}
	
	/**
	 * 重连
	 * @return
	 */
	public void reConnect(Message msg) {
		
		pool.execute(new Runnable() {
			
			public void run() {
				
				 group=new NioEventLoopGroup();
				
				try {
					Bootstrap b=new Bootstrap();
					 b.group(group)
					.channel(NioSocketChannel.class)
					.option(ChannelOption.TCP_NODELAY, true)
					.handler(new ChannelInitializer<SocketChannel>() {
						@Override
						protected void initChannel(SocketChannel arg0) throws Exception {
							
							arg0.pipeline().addLast(new Encoder());
							arg0.pipeline().addLast(new Decoder());
							
							arg0.pipeline().addLast("idleStateHandler", new IdleStateHandler(10,5,0));//心跳
							
							arg0.pipeline().addLast("HeartbeatHandler", new HeartbeatHandler());
							
							arg0.pipeline().addLast(new ChannelInboundHandlerAdapter(){
								
								@Override
								public void channelActive(ChannelHandlerContext ctx)
										throws Exception {
										context=ctx;
								}
								
								@Override
								public void channelRead(ChannelHandlerContext ctx,
										Object msg) throws Exception {
									
									if (listener!=null) {listener.OnMessage(msg);}
									
								}
								
								@Override
								public void exceptionCaught(
										ChannelHandlerContext ctx, Throwable cause)
										throws Exception {
								}
								
							});
						}
					});
					
					try {
					b.connect(client.getIp(), client.getPort()).channel().closeFuture().sync();
					} catch (InterruptedException e) {
				}
			} finally {
				group.shutdownGracefully();
				if (listener!=null) {listener.OnBreak();}
			}	
		}
	});
		if (msg!=null) {
			sendContent(msg);
		}
	}

	
	/**
	 * 发送消息
	 * @param msg
	 */
	public void sendContent(final Message msg) {
		if (context==null) {
			pool.execute(new Runnable() {
				
				@Override
				public void run() {
					while (context==null) {
						//如果连接为空一直阻塞
					}
					sendContent(msg);
				}
			});
		}else if (!context.channel().isActive()) {
			//如果连接非活跃 抛出断开事件
			if (listener!=null) {
				listener.OnBreak();
			}
		}else {
			context.writeAndFlush(msg);
		}
	}
	
	/**
	 * 断开连接
	 */
	public void disConnect() {
		group.shutdownGracefully();
	}
	
	
class HeartbeatHandler extends ChannelDuplexHandler {
	
	private int size=0;
		
		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg)
				throws Exception {
			
			if (msg instanceof Message) {
				ctx.fireChannelRead(msg);
			}else if (msg instanceof ByteBuf) {
				size=0;
			}
		}
		
		@Override  
	    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)  
	            throws Exception {  
	        if (evt instanceof IdleStateEvent) {  
	            IdleStateEvent e = (IdleStateEvent) evt;  
	            if (e.state() == IdleState.READER_IDLE) {
	            	
	            	size++;
	            	
	            	if (size>2) {
						ctx.close();
						size=0;
					}
	            	System.out.println("read time out!");
	            }else if (e.state()==IdleState.WRITER_IDLE) {
	            	ByteBuf buf=Unpooled.buffer();
	            	buf.writeByte(0);
	            	buf.writeInt(0);
	            	ctx.writeAndFlush(buf);
					System.out.println("send heartbeat packet!");
				}else if (e.state()==IdleState.ALL_IDLE) {
					ctx.close();
					System.out.println("all time out!");
				}
	        }  
	    }  
	    }
}

